package com.icbc.dccsh.bolt;

import java.util.HashMap;
import java.util.Map;

import com.icbc.dccsh.util.Watcher;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import flexjson.SerializationType;
import flexjson.StringBuilderOutputHandler;

public class JsonBolt extends BaseBasicBolt {
	private flexjson.JSONSerializer _serializer = null;
	private Watcher _watcher = null;

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		this._serializer = new flexjson.JSONSerializer();
		this._watcher = Watcher.getWatcher(this.getClass());
	}

	public void execute(Tuple tuple, BasicOutputCollector collector) {
		this._watcher.count();
		HashMap map = (HashMap) tuple.getValue(0);
		String json = this._serializer.deepSerialize(map);
		if (tuple.getSourceStreamId().equals("nested_stream")) {
			collector.emit("nested_stream", new Values(json));
		} else {
			collector.emit("flatten_stream", new Values(json));
		}	
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
//		declarer.declare(new Fields("json"));
		declarer.declareStream("flatten_stream", new Fields("json"));
		declarer.declareStream("nested_stream", new Fields("json"));
	}
}
